package com.atguigu.bigdata.chapter11.table_api;

import com.atguigu.bigdata.bean.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

// 静态导入: 可以把一个类中的静态成员直接导入, 然后在当前类中可以直接使用, 而不用添加类名

/**
 * @Author lzc
 * @Date 2022/9/9 14:09
 */
public class Flink01_Tabale_BaseUse_1 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        DataStreamSource<WaterSensor> stream = env.fromElements(
            new WaterSensor("sensor_1", 1000L, 10),
            new WaterSensor("sensor_2", 2000L, 20),
            new WaterSensor("sensor_1", 3000L, 30),
            new WaterSensor("sensor_1", 4000L, 40),
            new WaterSensor("sensor_2", 6000L, 50),
            new WaterSensor("sensor_1", 7000L, 60),
            new WaterSensor("sensor_1", 8000L, 80)
        );
        
        // 1. 先去创建表的执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        // 2. 通过表的执行环境把流转成一个动态表
        Table table = tEnv.fromDataStream(stream);
        
        // $ 其实是一个方法名, 因为是静态方法, 使用静态导入的方式, 可以直接使用
        Table result = table
            .where($("id").isEqual("sensor_1"))
            .select($("id"), $("ts"), $("vc"));
        
        // 4. 把表转成流, 打印
        DataStream<WaterSensor> resultStream = tEnv.toAppendStream(result, WaterSensor.class);
        resultStream.print();
    
    
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
